package belf.migrate.plugin.gbase;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.APIErrorCode;
import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;

import belf.migrate.api.exception.DataTypeMappingNotExistException;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.taskconf.TaskConf;
import belf.migrate.api.util.JdbcUrlUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;


import java.sql.*;
import java.util.*;

/**
 * 南大通用GBase作用目标端需要实现的数据库查询方法。
 * GBase 8s中只有database，没有schema。表名和视图名不能有引号。注意表名前缀写法：
 * <ul>
 *     <li>"owner".table：前缀是owner账号，用双引号，连接符是“.”</li>
 *     <li>database：tabel：前缀是数据库名称，连接符是“:”</li>
 * </ul>
 * GBase中的数据库连接已经带上了数据库名称，Catalog各种接口的方法实现中，database参数和schema其实都无效
 */
@Slf4j
public class GBaseCatalog extends AbstractSinkCatalog {
    private static final String QUOTE = "";
    private static final List<String> EXCLUDED_SCHEMAS =
            Collections.unmodifiableList(
                    Arrays.asList("GBASEDBT", "SYS", "SYSADMIN", "SYSMASTER", "SYSUSER", "SYSUTILS"));

    private static final String[][] DEFAULT_BEGIN_END_BLOCK = new String[][]{
            {
                    SQLConstants.BLOCK_BEGIN,
                    SQLConstants.BLOCK_END
            }
    };
    /**
     * GBase的jdbc url采用这种格式：gbasedbt-sqli://192.168.0.108:19088/testdb:GBASEDBTSERVER=gbase01
     * @param username
     * @param pwd
     * @param urlInfo
     * @param defaultSchema
     */
    public GBaseCatalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.GBASE_8S, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getListDatabaseSQL() {
        return "SELECT NAME FROM SYSMASTER:SYSDATABASES";
    }

    @Override
    public String getListTableSQL(String database, String schema) {
        return "SELECT TABLE_NAME FROM USER_TABLES";
    }

    @Override
    public String getListProcedureSQL(String database) {
//        return String.format("SELECT * FROM %s:SYSPROCEDURES WHERE mode not IN ('d','o','r') AND ISPROC='f'", database);
        return "select object_name as PROCEDURE_NAME from USER_PROCEDURES WHERE OBJECT_TYPE='PROCEDURE'";
    }

    @Override
    public String getListFunctionSQL(String database) {
//        return String.format("SELECT * FROM %s:SYSPROCEDURES WHERE mode not IN ('d','o','r') AND ISPROC='t'", database);
        return "select object_name as FUNCTION_NAME from USER_PROCEDURES WHERE OBJECT_TYPE='FUNCTION'";
    }

    @Override
    public String getListTriggerSQL(String database) {
//        return String.format("SELECT TRIGNAME AS TRIGGER_NAME FROM %s:SYSTRIGGERS", database);
        return "SELECT TRIGGER_NAME FROM USER_TRIGGERS";
    }

    @Override
    public String getListSequenceSQL(String database) {
        return "select SEQUENCE_NAME from USER_SEQUENCES";
    }

    @Override
    public String getListSynonymSQL(String database) {
        return "select tabname as SYNONYM_NAME from SYSTABLES where tabtype='S'";
    }

    /**
     * 对GBase数据库进行查询时，因为JDBC URL中已经限定了database，只需要指定表名即可
     * @param tablePath
     * @return
     */
    @Override
    public String getSelectColumnsSql(TablePath tablePath) {
        return "SELECT * FROM " + tablePath.getTableName();
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return String.format(
                "SELECT count(*) FROM %s LIMIT 1;",
                tablePath.getTableName());
    }

    @Override
    public String getDropTableSql(TablePath tablePath) {
        return "DROP TABLE " + tablePath.getTableName() + " CASCADE";
    }

    @Override
    public List<TableInfoModel> listTablesInfo(String databaseName, String schemaName) throws CatalogException, DatabaseNotExistException {
        try (PreparedStatement ps =
                     getConnection(defaultUrl)
                             .prepareStatement(getListTableSQL(databaseName, schemaName));
             ResultSet rs = ps.executeQuery()) {

            List<TableInfoModel> tables = new ArrayList<>();
            while (rs.next()) {
                if (EXCLUDED_SCHEMAS.contains(rs.getString(1))) {
                    continue;
                }

                TableInfoModel table = new TableInfoModel();
                table.setTableName(rs.getString("TABLE_NAME"));
                tables.add(table);
            }

            return tables;
        } catch (Exception e) {
            throw new CatalogException(
                    String.format("Failed listing table in catalog %s", catalogName), e);
        }
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getTableName();
    }

    @Override
    public String getQuote() {
        return QUOTE;
    }

    @Override
    public String getCreateTableSql(TablePath tablePath, CatalogTable table, TaskConf taskConf) throws DataTypeMappingNotExistException {
        return GBaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.GBASE_8S,
                        taskConf)
                .createTable();
    }

    @Override
    public List<String> getCreateIndexSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        return GBaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.GBASE_8S,
                        taskConf)
                .createIndexes();

    }

    @Override
    public List<String> getCreateForeignSqls(TablePath tablePath, CatalogTable table, TaskConf taskConf) {
        return GBaseCreateTableSqlBuilder.builder(tablePath,
                        table.getTableSchema(),
                        table.getCatalogName() + "-" + DatabaseIdentifier.GBASE_8S,
                        taskConf)
                .createForeigns();
    }
/*
    public void autoIncreamentOff(TablePath tablePath, Connection connection) {
        try {
//            String setSql = "set IDENTITY_INSERT " + tablePath.getFullName() + " ON";
            String setSql = "set IDENTITY_INSERT " + getFullTableName(tablePath) + " ON";
            log.info("Open IDENTITY_INSERT:" + setSql);
            Statement setStmt = connection.createStatement();
            setStmt.execute(setSql);
            setStmt.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    public void autoIncreamentOn(TablePath tablePath, Connection connection) {
        try {
//            String setSql = "set IDENTITY_INSERT " + tablePath.getFullName() + " OFF";
            String setSql = "set IDENTITY_INSERT " + getFullTableName(tablePath) + " OFF";
            log.info("Open IDENTITY_INSERT:" + setSql);
            Statement setStmt = connection.createStatement();
            setStmt.execute(setSql);
            setStmt.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }*/

    @Override
    public List<String> specialColumns() {
        return super.specialColumns();
    }

    @Override
    public List<String> bigDataColumns() {
        return List.of("LONGVARCHAR", "IMAGE", "LONGVARBINARY", "BLOB", "BFILE");
    }

    @Override
    protected String getColumnsSql(TablePath tablePath) {
        String tableName = caseSensitive() ? tablePath.getTableName().toUpperCase() : tablePath.getTableName();
        String sql = "select NULL as catalog_name, \n" +
                "                    \"schema_name\",COL.\"table_name\",\"schema_name\" as \"username\",\n" +
                "                    NAME as \"column_name\",\n" +
                "                    INFO2 as \"is_increment\",\n" +
                "                    COLID, \n" +
                "                    if(TYPE$ LIKE 'CLASS%',(select '\"'||owner||'\".\"'||object_name||'\"' from all_objects PKG_OBJ_INNER WHERE PKG_OBJ_INNER.OBJECT_TYPE = 'CLASS' and OBJECT_TYPE||''||OBJECT_ID=TYPE$\n" +
                "                ),TYPE$) as \"column_type\",\n" +
                "                    SF_GET_COLUMN_SIZE(TYPE$, CAST (LENGTH$ AS INT) ,CAST (SCALE AS INT)) \"data_length\", \n" +
                "                    SF_GET_COLUMN_SIZE(TYPE$, CAST (LENGTH$ AS INT) ,CAST (SCALE AS INT)) \"precision\",\n" +
                "                    SCALE as \"numeric_scale\", \n" +
                "                    (case when NULLABLE$='Y' then 1 else 0 end) \"nullable\",\n" +
                "                    DEFVAL \"data_default\",\n" +
                "                    if(au.column_name is null,0,1) \"is_primary_key\",\n" +
                "                    (select COMMENT$ from SYS.SYSCOLUMNCOMMENTS where SCHNAME=COL.\"schema_name\" and tvname = col.\"table_name\" and COLNAME=COL.NAME and TABLE_TYPE='TABLE') \"column_comment\"\n" +
                "                from (\n" +
                "                    select \n" +
                "                        D.name as \"schema_name\",C.name as \"table_name\",A.name,A.id,A.colid,CASE WHEN B.INFO1 IS NULL OR ((B.INFO1>>2) & 0x01)=0 THEN A.TYPE$ WHEN (B.INFO2 & 0xFF) = 0 THEN 'NUMBER' ELSE 'FLOAT' END AS TYPE$,CASE WHEN B.INFO1 IS NULL OR ((B.INFO1>>2) & 0x01)=0 THEN A.SCALE WHEN (B.INFO2 & 0xFF) = 0 THEN 0 ELSE 129 END AS SCALE,   CASE WHEN B.INFO1 IS NULL OR ((B.INFO1>>2) & 0x01)=0 THEN A.LENGTH$ ELSE (B.INFO2 & 0xFF) END AS LENGTH$,A.NULLABLE$,A.DEFVAL,A.info1,A.INFO2 \n" +
                "                    from SYS.SYSCOLUMNS A\n" +
                "                    LEFT JOIN SYS.SYSCOLINFOS B on A.ID=B.ID AND A.COLID=B.COLID \n" +
                "                    LEFT JOIN SYS.SYSOBJECTS C ON A.ID=C.ID\n" +
                "                    LEFT JOIN SYS.SYSOBJECTS D ON C.schid=D.ID\n" +
                "                    where A.ID in (\n" + "                        select id from SYS.SYSOBJECTS where";
        if (StringUtils.isNotEmpty(tableName)) {
            sql = sql + " NAME in ('" + tableName + "') and";
        }
        sql  = sql + "  schid in (select ID from SYS.SYSOBJECTS where TYPE$='SCH' and name = '" + tablePath.getRightDatabaseName() + "') and subtype$ in ('UTAB','STAB')\n" +
                "                    )  \n" +
                "                ) COL left join (select b.* from \n" +
                "                all_constraints a,all_cons_columns b \n" +
                "            where a.owner=b.owner and a.table_name=b.table_name and a.constraint_name=b.constraint_name and a.constraint_type='P') au on COL.\"schema_name\" = au.owner and COL.\"table_name\" = au.table_name and COL.NAME=au.column_name order by colid";
        return sql;
    }

    @Override
    public String getListViewsSQL(String databaseName) {
        return "select object_name as VIEW_NAME from USER_OBJECTS where object_type='VIEW'";
    }

    @Override
    protected String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return "select object_name as VIEW_NAME from USER_OBJECTS where object_type='VIEW'";
    }

    /**
     * GBase支持的自定义类型有限制，目前仅仅支持“CREATE ROW TYPE”命令创建的ROW类型
     * @param defaultDatabase
     * @return
     */
    @Override
    public String getListTypeSQL(String defaultDatabase) {
        return "select name as TYPE_NAME from SYSXTDTYPES where mode='R'";
    }

    @Override
    protected String getListIndexSQL(String defaultDatabase) {
        return "select INDEX_NAME,TABLE_NAME,INDEX_TYPE from USER_INDEXES";
    }

    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
                select c.constraint_name,ucc.column_name,ucc.position
               from USER_CONSTRAINTS c
               left join USER_CONS_COLUMNS ucc on c.constraint_name=ucc.constraint_name
               where c.table_name='%s'
               and c.constraint_type='U'
              order by ucc.position
              """,
                tablePath.getTableName());
        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("constraint_name");
                    String column = rs.getString("column_name");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        String sql = String.format("""
               select c.constraint_name,c.table_name,ucc.column_name,c.search_condition,c.delete_rule
               from USER_CONSTRAINTS c
               left join USER_CONS_COLUMNS ucc on c.constraint_name=ucc.constraint_name
               where c.table_name='EMPLOYEES' and c.constraint_type='C'
               """,
                tablePath.getTableName());
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);
        try(PreparedStatement ps = connection.prepareStatement(sql)) {
            try (ResultSet rs = ps.executeQuery()){
                while (rs.next()) {
                    String constraintName = rs.getString("constraint_name");
                    String searchCondition = rs.getString("search_condition");
                    String column = rs.getString("column_name");
                    String deleteRule = rs.getString("delete_rule");
                    CheckConstraint checkConstraint = new CheckConstraint(tablePath.getTableName(),
                            constraintName,
                            searchCondition,
                            column,
                            deleteRule);
                    checkConstraints.add(checkConstraint);
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }

    @Override
    public Object getColumnValueByType(ColumnMetaData columnMetaData, ResultSet resultSet, int index) throws SQLException {
        if (List.of("CLOB", "TEXT").contains(columnMetaData.getColumnTypeName())) {
            return resultSet.getString(index);
        }
        if ("BIT".equalsIgnoreCase(columnMetaData.getColumnTypeName())) {
            return resultSet.getInt(index);
        }
        if ("INT".equalsIgnoreCase(columnMetaData.getColumnTypeName())) {
            return resultSet.getInt(index);
        }
        return super.getColumnValueByType(columnMetaData, resultSet, index);
    }

    @Override
    public boolean dataEqualByType(String type, Object sourceData, Object sinkData) {
        if ("CHAR".equalsIgnoreCase(type)) {
            return sinkData.toString().trim().equals(sourceData.toString());
        }
        return super.dataEqualByType(type, sourceData, sinkData);
    }

    @Override
    public void setCurrentSchema(Connection connection, String schema) throws SQLException {
        String sql = String.format("SET SCHEMA %s", schema);
        connection.createStatement().execute(sql);
    }

    @Override
    public String[] getScriptDelimiters() {
        return SQLConstants.DEFAULT_SCRIPT_DELIMITER; //$NON-NLS-1$
    }

    @Override
    public String[][] getBlockBoundStrings() {
        return DEFAULT_BEGIN_END_BLOCK;
    }
}
